package data.platform.cassandra.internal.command;

import data.platform.cassandra.infra.persistence.mapping.RulePointEO;
import data.platform.cassandra.infra.persistence.mapping.RulePointKey;
import data.platform.cassandra.infra.persistence.repository.CassandraRulePointRepository;
import data.platform.common.service.command.RulePointCommandService;
import data.platform.common.stream.aggregates.RulePoint;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

@ConditionalOnBean(name = "cassandraConfig")
@Service
@Slf4j
@RequiredArgsConstructor
public class CassandraRulePointCommandServiceImpl implements RulePointCommandService {

    final CassandraRulePointRepository cassandraRulePointRepository;

    @Override
    public Mono<Integer> save(RulePoint rulePoint) {
        return Mono.just(rulePoint)
                .map(rp -> {
                    RulePointKey key = new RulePointKey();
                    key.setRuleId(rulePoint.getRuleId());
                    key.setPartition(rulePoint.getEventTime().toLocalDate());
                    key.setOffset(rulePoint.getEventTime().toLocalTime());

                    RulePointEO eo = new RulePointEO();
                    eo.setRulePointKey(key);
                    eo.setHitNumber(rulePoint.getHitNumber());
                    eo.setHitData(rulePoint.getHitData());
                    return eo;
                })
                .flatMap(eo -> cassandraRulePointRepository.insert(eo))
                .map(eo -> 1);
    }
}
